package com.innodealing.kafka.util;

import com.innodealing.kafka.exception.KafkaRegisterListenerException;
import com.innodealing.kafka.listener.HermesBatchOrderlyListener;
import com.innodealing.kafka.listener.KafkaBatchMessageListener;
import com.innodealing.onshore.datasync.plus.constant.KafkaConstants;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 动态 Kafka 监听器注册器
 *
 * @author Yinpenghai
 */
public class DynamicKafkaListenerRegistry {

    private static final int SINGLE_THREAD = 1;

    private final ConsumerFactory<Object, Object> consumerFactory;

    private final Map<String, Lock> registeringTopics = new ConcurrentHashMap<>();

    /**
     * 构造函数注入 ConsumerFactory
     *
     * @param consumerFactory Kafka 消费者工厂
     */
    public DynamicKafkaListenerRegistry(ConsumerFactory<Object, Object> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    private Map<String, ConcurrentMessageListenerContainer<Object, Object>> listenerContainers = new ConcurrentHashMap<>();

    /**
     * 注册新的 Kafka 监听器
     *
     * @param topic    需要监听的 topic
     * @param groupId  消费组 ID
     * @param listener 自定义的 Kafka 批量消息处理逻辑
     */
    public void registerKafkaListener(String topic, String groupId, HermesBatchOrderlyListener listener) {
        if (topic == null || topic.isEmpty()) {
            throw new KafkaRegisterListenerException("Topic cannot be empty.");
        }

        if (groupId == null || groupId.isEmpty()) {
            throw new KafkaRegisterListenerException("GroupId cannot be empty.");
        }

        final String key = getListenerKey(topic, groupId);
        Lock lock = registeringTopics.computeIfAbsent(key, k -> new ReentrantLock());

        // 使用锁防止重复注册
        lock.lock();
        try {
            if (listenerContainers.containsKey(key)) {
                throw new KafkaRegisterListenerException("Kafka listener for topic " + topic + " and groupId " + groupId + " already exists.");
            }

            final String realTopic = KafkaConstants.KAFKA_TOPIC_PREFIX + topic;
            ContainerProperties containerProps = new ContainerProperties(realTopic);

            containerProps.setGroupId(groupId);
            containerProps.setAckMode(ContainerProperties.AckMode.MANUAL);

            ConcurrentMessageListenerContainer<Object, Object> container =
                    new ConcurrentMessageListenerContainer<>(consumerFactory, containerProps);

            container.setupMessageListener(new KafkaBatchMessageListener(listener));

            container.setConcurrency(SINGLE_THREAD);
            container.start();

            listenerContainers.put(key, container);

        } finally {
            // 注册完成后，移除锁
            registeringTopics.remove(key);
            lock.unlock();
        }
    }

    /**
     * 根据 topic 和 groupId 生成唯一键
     */
    private String getListenerKey(String topic, String groupId) {
        return topic + "-" + groupId;
    }

    /**
     * 取消 Kafka 监听器
     *
     * @param topic 需要取消监听的 topic
     * @param groupId 消费组 ID
     */
    public void removeKafkaListener(String topic, String groupId) {
        if (topic == null || topic.isEmpty()) {
            throw new KafkaRegisterListenerException("Topic cannot be empty.");
        }
        final String key = getListenerKey(topic, groupId);

        ConcurrentMessageListenerContainer<Object, Object> container = listenerContainers.remove(key);
        if (container != null) {
            container.stop();
        } else {
            throw new KafkaRegisterListenerException("Kafka listener for topic " + topic + " not found.");
        }
    }
}
